package com.nucarf.yc.sink

import com.nucarf.yc.bean.ChangeOrderLog
import org.apache.flink.annotation.Public
import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.functions.sink.{RichSinkFunction, SinkFunction}

import java.sql.{Connection, DriverManager, PreparedStatement}

@Public
class JDBCSink extends RichSinkFunction[ChangeOrderLog]{

  var conn: Connection = _
  var insertStmt: PreparedStatement = _
  var updateStmt: PreparedStatement = _

  // open 主要是创建连接
  override def open(parameters: Configuration): Unit = {
    super.open(parameters)

    conn = DriverManager.getConnection("jdbc:mysql://cn-zz-bgp-1.natfrp.cloud:23576/king?useSSL=false", "root", "111111")
    insertStmt = conn.prepareStatement("INSERT INTO change_order_log (id, ts, table_name, database_name, operation, create_time, json_string) VALUES (?, ?, ?, ?, ?, ?, ?)")
//    updateStmt = conn.prepareStatement("UPDATE temperatures SET temp = ? WHERE sensor = ?")
  }
  // 调用连接，执行sql
  override def invoke(changeOrderLog: ChangeOrderLog, context: SinkFunction.Context[_]): Unit = {

    insertStmt.setString(1, changeOrderLog.id)
    insertStmt.setLong(2, changeOrderLog.ts)
    insertStmt.setString(3, changeOrderLog.tableName)
    insertStmt.setString(4, changeOrderLog.databaseName)
    insertStmt.setString(5, changeOrderLog.operation)
    insertStmt.setString(6, changeOrderLog.createTime)
    insertStmt.setString(7, changeOrderLog.jsonString)
    insertStmt.execute()
  }

  override def close(): Unit = {
    insertStmt.close()
    updateStmt.close()
    conn.close()
  }

}
